/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */


package org.apache.flink.api.common.eventtime;

import org.apache.flink.annotation.Public;

import java.time.Duration;

import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;

/**
 *
 * 类的加载遵守双亲委派模型。
 *  人话： 如果加载了两个同名(签名)的类，有先加载你自己写的。不加载引入的包下的。
 */
@Public
public class BoundedOutOfOrdernessWatermarks<T> implements WatermarkGenerator<T> {

    /** The maximum timestamp encountered so far. */
    private long maxTimestamp;

    /** The maximum out-of-orderness that this watermark generator assumes. */
    private final long outOfOrdernessMillis;

    /**
     * Creates a new watermark generator with the given out-of-orderness bound.
     *
     * @param maxOutOfOrderness The bound for the out-of-orderness of the event timestamps.
     */
    public BoundedOutOfOrdernessWatermarks(Duration maxOutOfOrderness) {
        checkNotNull(maxOutOfOrderness, "maxOutOfOrderness");
        checkArgument(!maxOutOfOrderness.isNegative(), "maxOutOfOrderness cannot be negative");

        this.outOfOrdernessMillis = maxOutOfOrderness.toMillis();

        // start so that our lowest watermark would be Long.MIN_VALUE.
        this.maxTimestamp = Long.MIN_VALUE + outOfOrdernessMillis + 1;
    }

    // ------------------------------------------------------------------------

    /*
            每来一条数据，就调用一次。
                更新收到的数据中的最大 时间属性。

                eventTimestamp： 从当前的T event 中提取的时间属性
                maxTimestamp： 当前算子收到的所有数据中最大的时间属性
                                默认this.maxTimestamp = Long.MIN_VALUE + outOfOrdernessMillis(0) + 1;

     */
    @Override
    public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {

        // 更新收到的数据中的最大 时间属性。 保证水印用于基于最大的时间属性制作，不会递减
        maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
        //System.out.println("收到了:"+event+",收到的最大的时间属性是:"+maxTimestamp);

        Watermark watermark = new Watermark(maxTimestamp - outOfOrdernessMillis - 1);

        //System.out.println(Thread.currentThread().getName() +"发送:"+watermark);
        //发送到下游
        output.emitWatermark(watermark);
    }

    /*
            制作水印，周期性发送。
                默认是200ms，可以自己调整。
     */
    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
       /* Watermark watermark = new Watermark(maxTimestamp - outOfOrdernessMillis - 1);

        System.out.println(Thread.currentThread().getName() +"发送:"+watermark);
        //发送到下游
        output.emitWatermark(watermark);*/
    }
}
